[2025-10-12] Kafka

🦥 본문

Motivation

Log Data

• 종류
  1. 사용자 활동 이벤트 : 로그인, 좋아요, 클릭…
  2. 운영 지표 : 콜 스택, 오류, 지연 시간, 자원 사용률
• 인터넷 애플리케이션 동향으로 로그가 생산 데이터의 일부가 됨. 다음과 같이 사용
  1. 검색 정확도
  2. 추천 시스템
  3. 광고 타게팅
  4. 보안 애플리케이션 : 스팸 차단
  5. 뉴스 피드

→ 데이터보다 볼륨이 커짐

기존의 한계

생산 서버에서 물리적으로 로그 파일 스크랩하는 방식에서 로그 수집기를 개발.

1. 기능 불일치 : 전달 보장 기능에 중점을 둠
   • 몇 개의 페이지 뷰 이벤트가 손실되는 것은 치명적이지 않음
   • 기존의 수집기는 트랜젝션 기능이나 소비된 후 승인 되는 것을 허용.

   → 과도한 기능으로 복잡성만 증가.

2. 처리량 부족
3. 분산 지원 미흡 : 메시지를 여러 머신에 분할하고 저장하는 쉬운 방법이 없음
4. 메시지 큐 가용성 문제 : 메시지의 즉각적인 소비 가정하여 메시지 큐 크기가 작음

특수 로그 수집기 한계

1. 오프라인 소비하도록 구축
2. 분 단위 파일과 같이 불필요하게 구현 사항을 소비자에게 노출

3. push 모델을 사용 : 브로커가 소비자에게 데이터를 전달하는 방법

   → pull 모델이 메시지 과부하를 방지, rewind.

Kafka

분산되고 확장 가능하며 높은 처리량을 제공을 제공하고 messaging system과 유사한 API를 제공하여 로그 이벤트를 실시간으로 소비 가능.

온라인/오프라인 소비 모두에 대해 단 하나의 소프트웨어를 사용하여 단순화

• 오프라인 소비 : 데이터를 나중에 소비하기 위해 저장

개념

• topic : 특정 유형의 메시지 스트림을 정의.
• producer : 프로듀서가 토픽에 메시지를 발행
• broker : 발행된 메시지가 저장되는 서버 집합.

• consumer : 브로커로 부터 한 개 혹은 여러 개의 topic을 구독할 수 있음. 데이터를 pull하여 소비. 여러 consumer가 토픽의 모든 메시지 복사본 하나를 공동으로 소비하는 point-to-point 전달 모델.

  consumer는 토픽의 자체 복사본을 검색하는 발행/구독 모델 지원

  • 구독

    consumer는 토픽에 대해 하나 이상의 메시지 스트림을 생성하고 발행된 메시지는 하위 스트림으로 균등하게 분배한다.

    각 메시지 스트림은 지속적으로 생성되는 메시지 스트림에 대한 iterator 인터페이스를 제공. iterator는 절대 종료되지 않음. 소비하는 메시지가 없으면 iterator는 새 메시지가 발행될 때까지 block 됨.

    • iterator interface : 컬렉션(집합체)의 내부 구조를 노출하지 않으면서 그 요소들을 순차적으로 접근(순회)할 수 있도록 표준화된 방법을 제공하는 인터페이스
• EX)

 producer = new Producer(…); 
  message = new Message(“test message str”.getBytes()); 
  set = new MessageSet(message); 
  producer.send(“topic1”, set); 

streams[] = Consumer.createMessageStreams(“topic1”, 1) 
  for (message : streams[0]) { 
    bytes = message.payload(); 
    // do something with the bytes  
  } 

Architecture

여러 브로커로 구성된 클러스터

• 로드 밸런싱을 위해 토픽은 여러 파티션으로 나뉨.
  • topic1/part1과 topic/part2 으로 2개의 파티션으로 나뉨
• 각 브로커는 하나 이상의 파티션을 저장
• 여러 프로듀서와 컨슈머가 동시에 메시지를 발행하고 검색

효율성 (단일 파티션 내에서)

• 간단한 저장소
  1. 로그와 파티션
     • 토픽의 각 파티션은 논리적 로그.
     • 로그는 동일한 크기. 세그먼트 파일들로 구현
  2. 메시지 발행
     • 프로듀서가 메시지를 발행할 때마다 브로커는 마지막 세그먼트 파일에 추가
     • 순차적으로 쓰기로 수행. 임의 쓰기에 비해 빠름
  3. 디스크 플러싱
     • 구성 가능한 메시지가 쌓이는 경우
     • 특정 시간이 경과한 경우
     • 플러싱 이후에 consumer에 메시지가 노출됨.
  4. 메시지 ID

     • 명시적 메시지 ID X 대신 논리적 오프셋으로 주소 지정

       → 랜덤 접근 인덱스 구조를 유지 관리하는 오버헤드를 피함

     • 다음 메시지 ID(=오프셋)는 (현재 ID + 메시지 길이)

  5. 메시지 소비
  • consumer가 특정 파티션에서 순차적으로 메시지 소비
  • 특정 메시지를 승인했다는 것은 이전 메시지를 모두 수신했다는 뜻.
  • 애플리케이션이 소비할 데이터 버퍼를 준비하기 위해 브로커에 비동기 pull 요청을 보냄
    • pull 요청 : 소비가 시작되는 메시지의 오프셋과 허용 가능한 바이트 수 포함 1. 오프셋 관리

    

    • 각 브로커는 모든 세그먼트 파일의 첫번째 메시지 오프셋을 포함하는 오프셋의 정렬된 리스트를 메모리에 유지.
    • 검색을 통해 consumer에게 보냄. consumer는 다음에 소비할 메시지의 오프셋을 계산하고 다음 pull 요청에 사용
• 효율적인 전송
  1. 배치 처리
     • 프로듀서 : 단일 전송 요청으로 메시지 세트를 브로커에 제출
     • 컨슈머 : 컨슈머 API는 한 번에 하나의 메시지 처리. pull 요청으로 여러 메시지를 한 번에 검색

  2. 운영 체제 기능 활용

     메시지를 메모리에 캐싱하는 것을 피하지만 기본 파일 시스템 페이지 캐시에 의존

     • 장점
       • 이중 버퍼링 회피 : Kafka 프로세스 메모리와 OS 페이지 캐시에 모두 저장하지 않아 메모리 사용의 효율적
       • 브로커 프로세스가 재시작되어도 운영체제의 페이지 캐시가 유지되어 warm cache 유지
       • 가비지 컬랙션 오버헤드 감소

     프로듀서와 컨슈머 모두 세그먼트 파일에 순차적이고 작은 양 접근하여 일반적인 캐싱 휴리스틱(write-through나 read-ahead)이 효과적. 데이터 크기에 선형적으로 비례하여 성능을 보임

  3. Zero-Copy

     하나의 메시지가 여러 컨슈머에게 여러 번 전송하는 것을 최적화

     • 일반적인 전송 과정
       1. 저장 매체 → OS의 페이지 캐시로 데이터 읽기
       2. 페이지 캐시 → 애플리케이션 버퍼로 복사
       3. 애플리케이션 버퍼 → 다른 커널 버퍼로 복사
       4. 커널 버퍼 → 소켓으로 전송

     → sendfile API를 활용하여 파일 채널에서 소켓 채널로 데이터를 직접 전송하여 2~3단계의 두 번의 복사화 한 번의 시스템 콜을 회피

• Stateless broker
  • 컨슈머의 소비 정보를 컨슈머가 관리하여 오버헤드를 낮추지만 어떤 구독자가 메시지를 모두 소비했는 지 모르기 때문에 메시지 삭제 시점을 결정하는 게 어려움

  • Time-based retention policy

    시간 기반 SLA 보존 정책. 특정 기간(약 7일)보다 오래되면 자동으로 삭제

    • 오프라인 컨슈머도 대체로 실시간~일별로 소비를 완료하여 잘 작동.
    • 누적 데이터의 크기에 따라 성능 저하가 이루어지지 않음

  • Rewind & Re-consume

    이전 오프셋을 rewind back하여 데이터를 다시 re-consume 가능

    • 사례
      1. 컨슈머 애플리케이션 로직 오류가 발생할 경우 재처리
      2. 주기적으로 플러싱 하는 경우 충돌이 발생하면 데이터가 유실될 수 있음. 플러싱되지 않은 메시지 중 가장 작은 오프셋을 체크포인트 삼아 재소비 가능

    → pull 모델의 이점

Distrinuted Coordination (분산 조정)

1. consumer group
   • 공동 소비 : 컨슈머 그룹은 하나 이상의 컨슈머로 구성. 구독된 topic set을 공동으로 소비
   • 단일 전달 : 각 메시지는 그룹 내의 단 하나의 컨슈머에게 전달
   • 독립적 소비 : 서로 다른 컨슈머 그룹은 구독된 메시지 전체를 각각 독립적으로 소비
2. 병렬 처리의 최소 단위
   • 토픽 내의 파티션을 병렬 처리의 최소 단위로 만듦
   • 한 파티션의 모든 메시지가 단일 컨슈머에 의해서만 소비
     • 여러 컨슈머가 소비하면 동기화 오버헤드 필요하지만 Kafka는 소비 프로세스가 재조정할 때만 조정하고 매우 드묾
   • over partitioning : 부하가 균형을 이루려면 토픽의 파티션 수가 각 그룹의 컨슈머 수보다 훨씬 많아야 함
3. 분산 조정 매커니즘 : Zookeeper
   • 마스터 노드 부재 : 마스터 노드를 제거하여 마스터 장애 가능성을 제거. 컨슈머들이 자체 조정하는 매커니즘을 사용. → 복잡성을 줄임

   • 작업

     주키퍼 작업	목적	레지스트리 경로
     변동 감지	브로커 및 컨슈머의 추가 및 제거 감지.	브로커 레지스트리 (Broker Registry), 컨슈머 레지스트리 (Consumer Registry)
     재조정 트리거	브로커/컨슈머 변동 시 각 컨슈머에게 재분배(rebalance) 프로세스 시작을 알림.	브로커 및 컨슈머 레지스트리에 와처(Watcher) 등록.
     관계 및 오프셋 유지	컨슈머 그룹과 파티션 간의 소비 관계 및 마지막 소비 오프셋 추적.	소유권 레지스트리 (Ownership Registry), 오프셋 레지스트리 (Offset Registry)

   • 특징
     • 파일 시스템과 유사한 API(경로 생성, 경로 값 설정/읽기 삭제 등)를 제공
     • 와처 등록 : 경로에 와처를 등록하여 경로의 자식 또는 값이 변경될 때 알림 → 브로커 세트 또는 컨슈머 그룹의 변경이 발생할 때마다 알림을 받음
     • 경로를 임시로 생성 가능 → 프로세스가 실패하면 소유권 자동 해제
     • 데이터를 여러 서버에 복제하여 신뢰성과 가용성 제공
   • 저장 정보
     • 브로커/컨슈머 레지스트리 : 각 브로커/컨슈머 시작시 Zookeeper에 저장
       • 브로커 레지스트리 : 브로커의 호스트명, 포트, 토픽 및 파티션 정보.
       • 컨슈머 레지스트리 : 컨슈머 그룹과 구독 토픽 정보.
     • 그룹 별 등록
       • 소유권 레지스트리 : 구독된 파티션별로 소비중인 컨슈머 ID 저장
       • 오프셋 레지스트리 : 마지막으로 소비된 메시지의 오프셋 저장. 영구적. 나머지는 임시
   • 장애 처리
     • 브로커가 실패하면 브로커 등록에서 자동으로 경로 제거
     • 컨슈머 실패시 컨슈머 등록에서 해당 항목을 읽고 소유권 등록에서 소유한 모든 파티션을 잃음

   • 재분배

     컨슈머가 시작되거나 변동을 통해 알림을 받으면 재분배 프로세스 시작

     1. 정보 수집 : 주키퍼에서 토픽의 사용가능한 파티션 집합과 구독 중인 컨슈머 집합을 읽어옴
     2. 파티션 분배 : 사용 가능한 파티션 집합을 구독 중인 컨슈머 집합 개의 덩어로로 나누고 하나의 덩어리를 소유
     3. 소유권 등록 : 컨슈머는 소유권 레지스트리에 ID를 등록
     4. 소비 시작 : 오프셋 레지스트리에 저장된 오프셋부터 데이터를 풀
     5. 오프셋 업데이트 : 주기적으로 오프셋 레지스트리에 업데이트
        • 재분배 충돌 처리 : 재분배 알림이 컨슈머 마다 약간 다른 시간에 와서 소유권 충돌이 발생할 때, 모든 파티션을 해제하고 재조정 프로세스를 재시도
        • 새 컨슈머 그룹 처리 : 컨슈머 그룹이 생성되어 오프셋 레지스트리에 정보가 없는 경우 설정에 따라 처음부터 끝까지 소비 시작

전달 보장

1. 최소 1회 전달
   • 한 번 이상 전달. 유실되지는 않음.
   • 중복 상황 : 컨슈머 프로세스가 깔끔하게 종료되지 않고 충돌하는 경우 인계 받는새 컨슈머는 ZooKeeper에 마지막으로 커밋된 오프셋 이후의 메시지 중 일부를 중복 해서 받음
   • 중복 제거 : 중복에 민감한 경우 컨슈머에서 중복 제거 로직을 추가해야 함. 오프셋의 고유키 사용
2. 메시지 순서 보장
   • 단일 파티션에서 메시지가 컨슈머에게 전달되는 순서 보장.
   • 다른 파티션에서 오는 메시지 사이의 순서는 보장 X
3. 무결성
   • 로그 손상 방지
     1. 로그의 각 메시지에 CRC(Cyclic Redundancy Check) 값 저장.
     2. 브로커에 I/O 오류가 발생하면 카프카는 CRC가 일치하지 않는 메시지를 제거하여 로그를 복구하는 프로세스 실행.
     3. 메시지 수준의 CRC를 통해 메시지가 생산/소비 후에 네트워크 오류도 확인 가능
4. 가용성
   • 상황
     • 브로커 다운 : 브로커에 저장된 아직 소비되지 않은 메시지는 사용 불가
     • 저장 시스템 손상 : 소비되지 않은 메시지는 영구적으로 유실
   • 향후 계획 : 이후에 여러 브로커에 중복 저장하여 해결할 계획

배포 및 활용

• 라이브 데이터센터 클러스터 : 사용자 대면 서비스가 실행되는 데이터 센터 내부에 위치
  • 흐름
    1. 프론트엔드 서비스가 로그 데이터를 batches(일괄 처리 방식)로 발행
    2. LB로 브로커에 균등하게 분배
    3. 동일한 데이터 센터 내의 온라인 컨슈머가 실시간으로 소비
• 오프라인 분석 클러스터 : 오프라인 분석 및 보고 지원
  1. 임베디드 컨슈머 세트를 실행하여 데이터를 복제
  2. Hadoop 및 데이터 웨어하우스로 데이터를 가져오는 데이터 로드 작업 실행
     • Hadoop : 대용량 데이터를 분산 처리하는 오픈소스 프레임워크입니다. 마치 여러 명의 일꾼이 함께 큰 짐을 나르는 것처럼, 데이터를 여러 컴퓨터에 나눠서 처리하여 효율성을 높입니다.
  3. 보고 작업 및 분석
  4. 프로토타이핑에 사용하여 임시 쿼리를 위해 원시 이벤트 스트림에 대해 간단한 스크립트 실행
  5. 종단 간 지연 시간은 약 10초
• 데이터 무결성
  • 감사 시스템
    • 메시지 메타 데이터 : 타임스탬프와 서버 이름 포함
    • 모니터링 이벤트 : 프로듀서는 고정된 시간 창 내에서 각 토픽에 대해 발행된 메시지 수를 기록하는 모니터링 이벤트를 주기적으로 생성
    • 컨슈머가 수신한 메시지 수를 모니터링 이벤트와 비교하여 검증
  • Hadoop 통합
    • hadoop 클러스터로 데이터를 로드하기 위해 카프카 입력 형식을 구현하여 데이터를 직접 읽을 수 있게 함
    • stateless 브로커와 클라이언트 측 메시지 오프셋 저장을 통해 작업 재시작시 중복이나 손실 없이 자연스럽게 로드
    • 데이터와 오프셋은 작업이 성공적으로 완료된 후에만 HDFS에 저장
• 메시지 직렬화
  • Avro 채택 : 직렬화 프로토콜 사용
    • 각 메시지 페이로드에 직렬화된 스키마 ID 저장
    • 경량 스키마 레지스트리 서비스 사용 → 스키마 ID를 실제 스키마에 매핑
    • 호환성 : 컨슈머는 메시지를 받을 때 스키마 레지스트리에서 스키마를 찾아 바이트를 객체로 디코딩 → 프로듀서-컨슈머 데이터 호환성을 강제